package com.cctc.rds.scheduler.server;

import com.cctc.rds.nrpc.api.rpc.RpcServerRouter;
import com.cctc.rds.nrpc.api.server.config.RpcServerConfig;
import com.cctc.rds.nrpc.core.server.NettyRpcServer;
import com.cctc.rds.raft.core.client.command.RaftCommandClient;
import com.cctc.rds.raft.core.log.LogService;
import com.cctc.rds.raft.core.log.commit.statemachine.StateMachine;
import com.cctc.rds.raft.core.log.impl.DefaultLogService;
import com.cctc.rds.raft.core.node.NodeIpPort;
import com.cctc.rds.raft.core.util.SignalManager;
import com.cctc.rds.raft.sdk.RaftStarter;
import com.cctc.rds.scheduler.server.config.ApplicationConfig;
import com.cctc.rds.scheduler.server.config.NettyConfig;
import com.cctc.rds.scheduler.server.config.helper.ApplicationConfigHolder;
import com.cctc.rds.scheduler.server.config.helper.YamlReader;
import com.cctc.rds.scheduler.server.raft.apply.TaskLogStateMachine;
import com.cctc.rds.scheduler.server.raft.helper.RaftCommandClientHolder;
import com.cctc.rds.scheduler.server.raft.rpc.TaskRpcServerRouter;
import com.cctc.rds.scheduler.server.raft.rpc.impl.TaskRpcResponseImpl;
import com.cctc.rds.scheduler.server.storage.Storage;
import com.cctc.rds.scheduler.server.storage.impl.LevelDBStorage;
import com.cctc.rds.scheduler.server.storage.leveldb.LevelDB;

import java.util.HashSet;
import java.util.Set;

/**
 * 任务延迟调度器服务端 [启动类]。
 */
public class SchedulerServerApplication {

    public static void main(String[] args) {
        String activeEnv = (args == null || args.length == 0) ? "dev" : args[0];
        String activeEnvConfigFile = "application-" + activeEnv + ".yaml";
        ApplicationConfig applicationConfig = new YamlReader<ApplicationConfig>(activeEnvConfigFile).loadConfig();
        ApplicationConfigHolder.holdApplicationConfig(applicationConfig);

        NettyConfig nettyConfig = applicationConfig.getNetty();
        int port = nettyConfig.getPort();
        Set<NodeIpPort> nodes = new HashSet<>(nettyConfig.getNodes());
        int nodeId = -1;
        for (NodeIpPort node : nodes) {
            if (node.getPort() == port) nodeId = node.getNodeId();
        }
        if (nodeId == -1) throw new RuntimeException("NettyConfig 没有正确的配置 Raft 节点 ID！！！");

        String storagePath = applicationConfig.getTask().getStoragePath();
        LevelDB.open(storagePath);
        Storage storage = new LevelDBStorage();

        LogService logService = new DefaultLogService();
        StateMachine stateMachine = new TaskLogStateMachine(storage);
        RaftCommandClient raftCommandClient = RaftStarter.start(nodeId, logService, stateMachine, nodes);
        RaftCommandClientHolder.holdRaftCommandClient(raftCommandClient);

        RpcServerConfig rpcServerConfig = RpcServerConfig.builder()
                .port(port)
                .workerThreads(nettyConfig.getWorkerThreads())
                .idleStateMs(nettyConfig.getIdleTimeoutMs())
                .build();
        RpcServerRouter rpcServerRouter = new TaskRpcServerRouter(new TaskRpcResponseImpl(storage));
        NettyRpcServer rpcServer = new NettyRpcServer(rpcServerRouter);
        rpcServer.start(rpcServerConfig);

        SignalManager.addFirst(signal -> rpcServer.stop());
        SignalManager.addFirst(raftCommandClient, storage);
    }

}
